package com.deep.flink;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author sw
 * @create 2023-06-30 11:26
 */
public class Test11 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        Table table = tableEnv.sqlQuery("select username, " +
                "    TUMBLE_END(ts,  NTERVAL '1' HOUR ) as endT, " +
                "    COUNT(url) as cnt " +
                " from `test5`" +
                " group by username," +
                "  TUMBLE(ts,  INTERVAL '1' HOUR ) ");
        table.printSchema();


        Table table2 = tableEnv.sqlQuery("select username, " +
                "    window_end as endT, " +
                "    COUNT(url) as cnt " +
                " from `test5` (" +
                "  TUMBLE( TABLE test5, DESCRIPTOR(ts), INTERVAL '1' HOUR) " +
                " ) " +
                " group by username, window_start, window_end ");
        table2.printSchema();


        env.execute();

    }
}
